package com.didispace;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.annotation.rxjava.EnableRxJavaProcessor;
import org.springframework.cloud.stream.annotation.rxjava.RxJavaProcessor;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.Transformer;
import org.springframework.messaging.handler.annotation.SendTo;

import java.text.SimpleDateFormat;
import java.util.Date;

// App1 中实现了对 input 输入通道的监听，并且在接收到消息之后，对消息做了一些简单加工，
// 然后通过 @SendTo 把处理方法返回的内容以消息的方式发送到 output 通道中  p364
// 消息反馈  p364
// @EnableRxJavaProcessor
// @EnableBinding(value = {Processor.class})
public class App1 {

    private static Logger logger = LoggerFactory.getLogger(App1.class);

    // rxjava实现  p366  再次运行 App1 和 App2 就可以得到与消息反馈中的试验相同的结果
    @Bean
    public RxJavaProcessor<String,String> processor() {
        return inputStream -> inputStream.map(data -> {
            logger.info("Received: " + data);
            return data;
        // }).map(data -> String.valueOf("From Input Channel Return - " + data));
        // App1 在接收到 5 条消息之后才将处理结果返回给输出通道  p367
        }).buffer(5).map(data -> String.valueOf("From Input Channel Return - " + data));
    }

   // @StreamListener(Processor.INPUT)
   // @SendTo(Processor.OUTPUT)
   // public Object receiveFromInput(Date payload) {
   //     logger.info("Received: " + payload);
   //     return "From Input Channel Return - " + payload;
   // }

    // 原生实现  使用原生注解 @ServiceActivator 的 outputChannel 属性配置输出通道把返回结果发送给消息中间件  p365
   // @ServiceActivator(inputChannel= Processor.INPUT, outputChannel = Processor.OUTPUT)
   // public Object receiveFromInput(Object payload) {
   //     long payloadL = Long.parseLong((String) payload);
   //     Date payloadDate = new Date(payloadL*1000);
   //     logger.info("Received: " + payloadDate);
   //     return "From Input Channel Return - " + payloadDate;
   // }
}
